package com.central.scheduler.master.runner;

import com.central.scheduler.common.Constants;
import com.central.scheduler.common.enums.ExecutionStatus;
import com.central.scheduler.common.enums.StateEvent;
import com.central.scheduler.common.enums.StateEventType;
import com.central.scheduler.common.enums.TimeoutFlag;
import com.central.scheduler.common.thread.Stopper;
import com.central.scheduler.common.utils.DateUtils;
import com.central.scheduler.dao.model.ProcessInstance;
import com.central.scheduler.dao.model.TaskInstance;
import org.apache.hadoop.util.ThreadUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Tindy
 * @date 2021/11/4
 * @describe
 * 1. timeout check wheel
 * 2. dependent task check wheel
 */
public class StateWheelExecuteThread extends Thread {

    private static final Logger logger = LoggerFactory.getLogger(StateWheelExecuteThread.class);

    ConcurrentHashMap<Long, ProcessInstance> processInstanceCheckList;
    ConcurrentHashMap<Long, TaskInstance> taskInstanceCheckList;
    private ConcurrentHashMap<Long, WorkflowExecuteThread> processInstanceExecMaps;

    private int stateCheckIntervalSecs;
    public StateWheelExecuteThread(ConcurrentHashMap<Long, ProcessInstance> processInstances,
                                   ConcurrentHashMap<Long, TaskInstance> taskInstances,
                                   ConcurrentHashMap<Long, WorkflowExecuteThread> processInstanceExecMaps,
                                   int stateCheckIntervalSecs) {
        this.processInstanceCheckList = processInstances;
        this.taskInstanceCheckList = taskInstances;
        this.processInstanceExecMaps = processInstanceExecMaps;
        this.stateCheckIntervalSecs = stateCheckIntervalSecs;
    }

    @Override
    public void run() {

        logger.info("state wheel thread start");
        while (Stopper.isRunning()) {
            try {
                checkProcess();
                checkTask();
            } catch (Exception e) {
                logger.error("state wheel thread check error:", e);
            }
            ThreadUtil.sleepAtLeastIgnoreInterrupts(stateCheckIntervalSecs);
        }
    }
    private void checkTask() {
        if (taskInstanceCheckList.isEmpty()) {
            return;
        }

        for (TaskInstance taskInstance : this.taskInstanceCheckList.values()) {
            if (TimeoutFlag.OPEN == taskInstance.getTimeoutParam().getFlag()) {
                long timeRemain = DateUtils.getRemainTime(taskInstance.getStartTime(), taskInstance.getTimeoutParam().getInterval() * Constants.SEC_2_MINUTES_TIME_UNIT);
                if (0 <= timeRemain && processTimeout(taskInstance)) {
                    taskInstanceCheckList.remove(taskInstance.getId());
                    return;
                }
            }
            if (taskInstance.taskCanRetry() && taskInstance.retryTaskIntervalOverTime()) {
                processDependCheck(taskInstance);
                taskInstanceCheckList.remove(taskInstance.getId());
            }
            if (taskInstance.isSubProcess() || taskInstance.isDependTask()) {
                processDependCheck(taskInstance);
            }
        }
    }

    private void checkProcess() {
        if (processInstanceCheckList.isEmpty()) {
            return;
        }
        for (ProcessInstance processInstance : this.processInstanceCheckList.values()) {

            long timeRemain = DateUtils.getRemainTime(processInstance.getStartTime(), processInstance.getTimeout() * Constants.SEC_2_MINUTES_TIME_UNIT);
            if (0 <= timeRemain && processTimeout(processInstance)) {
                processInstanceCheckList.remove(processInstance.getId());
            }
        }
    }
    private void putEvent(StateEvent stateEvent) {

        if (!processInstanceExecMaps.containsKey(stateEvent.getProcessInstanceId())) {
            return;
        }
        WorkflowExecuteThread workflowExecuteThread = this.processInstanceExecMaps.get(stateEvent.getProcessInstanceId());
        workflowExecuteThread.addStateEvent(stateEvent);
    }

    private boolean processDependCheck(TaskInstance taskInstance) {
        StateEvent stateEvent = new StateEvent();
        stateEvent.setType(StateEventType.TASK_STATE_CHANGE);
        stateEvent.setProcessInstanceId(taskInstance.getProcessInstanceId());
        stateEvent.setTaskInstanceId(taskInstance.getId());
        stateEvent.setExecutionStatus(ExecutionStatus.RUNNING_EXECUTION);
        putEvent(stateEvent);
        return true;
    }

    private boolean processTimeout(TaskInstance taskInstance) {
        StateEvent stateEvent = new StateEvent();
        stateEvent.setType(StateEventType.TASK_TIMEOUT);
        stateEvent.setProcessInstanceId(taskInstance.getProcessInstanceId());
        stateEvent.setTaskInstanceId(taskInstance.getId());
        putEvent(stateEvent);
        return true;
    }

    private boolean processTimeout(ProcessInstance processInstance) {
        StateEvent stateEvent = new StateEvent();
        stateEvent.setType(StateEventType.PROCESS_TIMEOUT);
        stateEvent.setProcessInstanceId(processInstance.getId());
        putEvent(stateEvent);
        return true;
    }
}
